package dyyx.zb;

import java.util.List;
import java.util.Map;

import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;

import dyyx.util.TestDataUtil;

public class MapDataSetTest {

	public static void main(String[] args) throws Exception {

		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		// env.setParallelism(1);

		List<Map<String, Object>> list = TestDataUtil.getTestEventList2();

		DataSet<Map<String, Object>> dataset = env.fromCollection(list);

		dataset.groupBy("S_S_ACCTNO")
		.sortGroup("D_T_VB_EVENTOCCURTIME", Order.ASCENDING)
		.combineGroup(new MyGroupCombineFunction()).print();
		
		// This type (GenericType<java.util.HashMap>) cannot be used as key
		// org.apache.flink.api.common.InvalidProgramException: This type (GenericType<java.util.HashMap>) cannot be used as key.

	}

	private static class MyGroupCombineFunction
			implements GroupCombineFunction<Map<String, Object>, Map<String, Object>> {

		private static final long serialVersionUID = 1L;

		public void combine(Iterable<Map<String, Object>> values, Collector<Map<String, Object>> out) throws Exception {
			for (Map<String, Object> item : values) {
				out.collect(item);
			}
		}
	}

}
